/**
 * Copyright (C) 2012-2013 jumei, Inc.
 * DefaultServicePipeline.java
 * 2015年6月23日
 */
package org.goface.rpc.pipeline;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;

import java.util.LinkedList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.goface.rpc.Message;
import org.goface.rpc.MessageContext;
import org.goface.rpc.RPC;
import org.goface.rpc.invoker.ServiceInvoker;
import org.goface.rpc.net.ClientSideConnectionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.goface.rpc.conf.ServiceConfigure;
import com.goface.rpc.conf.ServiceConfigureCenter;
import com.google.common.util.concurrent.SettableFuture;

/**
 *
 * @author jm110
 */
public class DefaultServicePipeline implements ServicePipeLine {
    protected Logger LOG = LoggerFactory.getLogger(DefaultServicePipeline.class);
    protected ExecutorService executorService;

    protected LinkedList<ServiceHandlerContext> handlerContexts;

    public DefaultServicePipeline(ExecutorService executorService) {
        this();
        this.executorService = executorService;
    }

    public DefaultServicePipeline() {
        init();
        executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
    }

    private void init() {
        handlerContexts = new LinkedList<ServiceHandlerContext>();
        ServiceHandler headerHandler = new HeaderHandler();
        DefaultServiceHandlerContext headHandlerContext = new DefaultServiceHandlerContext("head", headerHandler);
        headHandlerContext.setPipeLine(this);

        ServiceHandler tailHandler = new TailHandler();
        DefaultServiceHandlerContext tailHandlerContext = new DefaultServiceHandlerContext("tail", tailHandler);
        tailHandlerContext.setPipeLine(this);
        handlerContexts.add(headHandlerContext);
        handlerContexts.add(tailHandlerContext);

        headHandlerContext.setNext(tailHandlerContext);
        tailHandlerContext.setPreviouse(headHandlerContext);

    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.goface.rpc.ServicePipeLine#fireSendRequest(org.goface.rpc.Message)
     */
    @Override
    public void fireSendRequest(MessageContext request) throws Throwable {
        ServiceHandlerContext context = handlerContexts.getFirst();
        context.getHandler().sendRequest(context, request);
    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.goface.rpc.ServicePipeLine#fireSendResponse(org.goface.rpc.Message)
     */
    @Override
    public void fireSendResponse(MessageContext response) throws Throwable {
        ServiceHandlerContext context = handlerContexts.getFirst();
        context.getHandler().sendResponse(context, response);

    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.goface.rpc.ServicePipeLine#fireReceiveRequest(org.goface.rpc.Message)
     */
    @Override
    public void fireReceiveRequest(MessageContext request) throws Throwable {
        ServiceHandlerContext context = handlerContexts.getLast();
        context.getHandler().receiveRequest(context, request);
    }

    /*
     * (non-Javadoc)
     *
     * @see
     * org.goface.rpc.ServicePipeLine#fireReceiveResponse(org.goface.rpc.Message
     * )
     */
    @Override
    public void fireReceiveResponse(MessageContext response) throws Throwable {
        ServiceHandlerContext context = handlerContexts.getLast();
        context.getHandler().receiveResponse(context, response);
    }

    /*
     * (non-Javadoc)
     *
     * @see org.goface.rpc.ServicePipeLine#addLast()
     */
    @Override
    public void addLast(ServiceHandler handler) {
        ServiceHandlerContext last = handlerContexts.removeLast();
        DefaultServiceHandlerContext context = new DefaultServiceHandlerContext(handler);
        context.setPipeLine(this);
        last.getPreviouse().setNext(context);
        context.setPreviouse(last.getPreviouse());
        context.setNext(last);
        last.setPreviouse(context);
        handlerContexts.add(context);
        handlerContexts.add(last);

    }

    /*
     * (non-Javadoc)
     *
     * @see org.goface.rpc.ServicePipeLine#addFirst()
     */
    @Override
    public void addFirst(ServiceHandler handler) {
        ServiceHandlerContext first = handlerContexts.removeFirst();
        DefaultServiceHandlerContext context = new DefaultServiceHandlerContext(handler);
        context.setPipeLine(this);
        context.setNext(first.getNext());
        context.setPreviouse(first);
        first.setNext(context);
        handlerContexts.addFirst(context);
        handlerContexts.addFirst(first);
    }

    protected class HeaderHandler implements ServiceHandler {
        protected Logger LOG = LoggerFactory.getLogger(DefaultServicePipeline.class);

        /*
         * (non-Javadoc)
         *
         * @see org.goface.rpc.ServiceHandler#sendRequest(org.goface.rpc.
         * ServiceHandlerContext, org.goface.rpc.Message)
         */
        @Override
        public void sendRequest(ServiceHandlerContext context, MessageContext messageCtx) throws Throwable {
            context.getNext().getHandler().sendRequest(context.getNext(), messageCtx);
        }

        /*
         * (non-Javadoc)
         *
         * @see org.goface.rpc.ServiceHandler#sendResponse(org.goface.rpc.
         * ServiceHandlerContext, org.goface.rpc.Message)
         */
        @Override
        public void sendResponse(ServiceHandlerContext context, MessageContext messageCtx) throws Throwable {
            messageCtx.getProperties().put(MessageContext.MESSAGE_TYPE, MessageContext.MESSAGE_TYPE_REPLY);
            context.getNext().getHandler().sendResponse(context.getNext(),messageCtx);
        }

        /*
         * (non-Javadoc)
         *
         * @see org.goface.rpc.ServiceHandler#receiveRequest(org.goface.rpc.
         * ServiceHandlerContext, org.goface.rpc.Message)
         */
        @Override
        public void receiveRequest(ServiceHandlerContext context, MessageContext messageCtx) throws Throwable {
            String serviceName = messageCtx.getMessage().getServiceName();
            String operator = messageCtx.getMessage().getOperator();
            Object[] args = messageCtx.getMessage().getArgs();
            Class<?> argClazz[] = null;
            if (args != null && args.length != 0) {
                argClazz = new Class<?>[args.length];
                int index = 0;
                for (Object arg : args) {
                    argClazz[index++] = arg.getClass();
                }
            }
            ServiceInvoker invoker = messageCtx.getServiceInvoker();
            Object retValue = null;
            try {
                retValue = invoker.invork(serviceName, operator, argClazz, args);
            } catch (Throwable th) {
                LOG.error("invork service failed", th);
                messageCtx.getProperties().put(MessageContext.MESSAGE_TYPE, MessageContext.MESSAGE_TYPE_REPLY);
                messageCtx.getMessage().setCause("service call exception:" + th.getMessage());
                context.getPipeLine().fireException(new RuntimeException("service call exception:" + th.getMessage()));
                return;
            }
            messageCtx.getMessage().setReply(retValue);
            context.getPipeLine().fireSendResponse(messageCtx);

        }

        /*
         * (non-Javadoc)
         *
         * @see org.goface.rpc.ServiceHandler#receiveResponse(org.goface.rpc.
         * ServiceHandlerContext, org.goface.rpc.Message)
         */
        @Override
        public void receiveResponse(ServiceHandlerContext context, MessageContext messageCtx) throws Throwable {

            boolean isAsync = messageCtx.isAsyncInvok();
            if (isAsync) {
                // TODO 切换线程进行回调

            } else {
                // 唤醒等待线程
                SettableFuture<Message> future = messageCtx.getRetFuture();
                future.set(messageCtx.getMessage());
            }
        }

        /*
         * (non-Javadoc)
         *
         * @see org.goface.rpc.ServiceHandler#onException(org.goface.rpc.
         * ServiceHandlerContext, org.goface.rpc.Message)
         */
        @Override
        public void onException(ServiceHandlerContext context, MessageContext messageCtx) {
            String cause = messageCtx.getMessage().getCause();
            RuntimeException ex = new RuntimeException(cause);
            boolean isAsync = messageCtx.isAsyncInvok();
            if (isAsync) {
                // TODO 切换线程进行回调

            } else {
                // 唤醒等待线程
                SettableFuture<Message> future = messageCtx.getRetFuture();
                future.setException(ex);
            }
        }

    }

    protected class TailHandler implements ServiceHandler {
        ClientSideConnectionManager connectionMgr = ClientSideConnectionManager.getInstance();

        /*
         * (non-Javadoc)
         *
         * @see org.goface.rpc.ServiceHandler#sendRequest(org.goface.rpc.
         * ServiceHandlerContext, org.goface.rpc.Message)
         */
        @Override
        public void sendRequest(ServiceHandlerContext context, MessageContext messageCtx) throws Throwable {
            Channel channel = connectionMgr.getOrCreateServerChannel(messageCtx.getServiceAddress(),
                    context.getPipeLine());
            SettableFuture<Message> retWaitFuture = SettableFuture.create();
            messageCtx.setRetFuture(retWaitFuture);
            ChannelFuture res = channel.writeAndFlush(messageCtx);
            if (messageCtx.isOneWay()) {
                res.get(10, TimeUnit.SECONDS);// 等待发送结果
                messageCtx.getMessage().setReply(null);// 复用
                retWaitFuture.set(messageCtx.getMessage());
                return;// oneway 调用
            }
            if (!messageCtx.isAsyncInvok()) {
                res.get(10, TimeUnit.SECONDS);// 等待发送结果
                Message message = retWaitFuture.get(60, TimeUnit.SECONDS);
                if (message.getCause() != null) {
                    // 服务调用异常
                    context.getPipeLine().fireException(new RuntimeException(message.getCause()));
                }
                context.getPipeLine().fireReceiveResponse(messageCtx);
            }
        }

        /*
         * (non-Javadoc)
         *
         * @see org.goface.rpc.ServiceHandler#sendResponse(org.goface.rpc.
         * ServiceHandlerContext, org.goface.rpc.Message)
         */
        @Override
        public void sendResponse(ServiceHandlerContext context, MessageContext messageCtx) throws Throwable {
            messageCtx.getClientChannel().write(messageCtx.getMessage());

        }

        /*
         * (non-Javadoc)
         *
         * @see org.goface.rpc.ServiceHandler#receiveRequest(org.goface.rpc.
         * ServiceHandlerContext, org.goface.rpc.Message)
         */
        @Override
        public void receiveRequest(ServiceHandlerContext handlerContext, MessageContext messageCtx) throws Throwable {
            // 切换线程
            final ServiceHandlerContext context = handlerContext;
            final MessageContext messageContext = messageCtx;
            context.getPipeLine().getExecutor().execute(new Runnable() {

                @Override
                public void run() {
                    if (context.getPreviouse() != null) {
                        try {
                            context.getPreviouse().getHandler().receiveRequest(context.getPreviouse(), messageContext);
                        } catch (Throwable th) {
                            messageContext.getProperties().put(MessageContext.MESSAGE_TYPE,
                                    MessageContext.MESSAGE_TYPE_REPLY);
                            messageContext.getMessage().setCause("server error:" + th.getMessage());
                            context.getPipeLine()
                                    .fireException(new RuntimeException("server error:" + th.getMessage()));
                        }
                    }
                }
            });

        }

        /*
         * (non-Javadoc)
         *
         * @see org.goface.rpc.ServiceHandler#receiveResponse(org.goface.rpc.
         * ServiceHandlerContext, org.goface.rpc.Message)
         */
        @Override
        public void receiveResponse(ServiceHandlerContext context, MessageContext messageCtx) throws Throwable {
            context.getPreviouse().getHandler().receiveResponse(context.getPreviouse(), messageCtx);

        }

        /*
         * (non-Javadoc)
         *
         * @see org.goface.rpc.ServiceHandler#onException(org.goface.rpc.
         * ServiceHandlerContext, org.goface.rpc.Message)
         */
        @Override
        public void onException(ServiceHandlerContext context, MessageContext message) {

        }

    }

    /*
     * (non-Javadoc)
     *
     * @see org.goface.rpc.ServicePipeLine#fireException(java.lang.Throwable)
     */
    @Override
    public void fireException(Throwable th) {
        // TODO Auto-generated method stub

    }

    @Override
    public ExecutorService getExecutor() {
        return this.executorService;
    }

    @Override
    public void desdroy() {
        if (executorService != null) {
            executorService.shutdownNow();
        }

    }

}
